package com.itheima.kafka.boot.listen;

import com.itheima.kafka.boot.stream.KafkaStreamListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.util.Arrays;

/**
 * @Description:
 * @Version: V1.0
 */
@Component
@Slf4j
public class StreamListener implements KafkaStreamListener<KStream<String,String>> {


    /**
     * 接收待处理消息topic
     * @return
     */
    @Override
    public String listenerTopic() {
        return  "itcast-topic-input";
    }

    /**
     * 处理成功后发送消息 topic
     * @return
     */
    @Override
    public String sendTopic() {
        return  "itcast-topic-output";
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
       return stream.flatMapValues((key, value)->{
           System.out.println("接收到的待处理的消息：" + value);
           return Arrays.asList(  value.split(" ") );
       })
               // value 分组
           .groupBy( (key,value) -> value)
           .windowedBy(TimeWindows.of(5000))
           .count()
           .toStream()
           .map((key,value)->{
               // key ： hello
               // value: 2
               return new KeyValue<>(key.key().toString(), value.toString());
           });

    }
}
